package com.bl.file.controller;

import com.bl.file.constants.CommonConstant;
import com.bl.file.util.UploadLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author yangxiaodong
 * @version 1.0
 * @description: TODO
 * @date 2025/7/2 10:24
 */
@RestController
@Slf4j
public class FileController {

    @Value("${file.upload-dir:temp}")
    private String uploadDir;

    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;


    @PostMapping(value = "/upload-chunk", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<String> uploadChunk(@RequestPart("file") FilePart chunk, @RequestParam("index") int index,
                                    @RequestParam("uploadId") String uploadId) throws IOException {
        Path tempDir = Paths.get(uploadDir, uploadId);
        if (!Files.exists(tempDir)) {
            Files.createDirectories(tempDir);
        }
        Path chunkPath = tempDir.resolve(index + CommonConstant.CHUNK_FILE_SUFFIX);
        return chunk.transferTo(chunkPath)
                .then(reactiveStringRedisTemplate.opsForSet().add(String.format(CommonConstant.UPLOAD_REDIS_KEY, uploadId), String.valueOf(index)))
                .thenReturn("Chunk " + index + " uploaded");
    }

    @GetMapping("/uploaded")
    public Mono<Set<Integer>> getUploadedChunks(@RequestParam String uploadId) {
        return reactiveStringRedisTemplate.opsForSet()
                .members(String.format(CommonConstant.UPLOAD_REDIS_KEY, uploadId))
                .map(Integer::valueOf)
                .collect(Collectors.toSet());
    }

    @PostMapping("/merge")
    public Mono<String> mergeChunks(
            @RequestParam("uploadId") String uploadId,
            @RequestParam("fileName") String fileName,
            @RequestParam("totalChunks") int totalChunks
    ) {
        Path tempDir = Paths.get(uploadDir, uploadId);
        Path finalFile = Paths.get(uploadDir, fileName);

        try (OutputStream outputStream = Files.newOutputStream(finalFile, StandardOpenOption.CREATE)) {
            for (int i = 0; i < totalChunks; i++) {
                Path chunkPath = tempDir.resolve(i + CommonConstant.CHUNK_FILE_SUFFIX);
                Files.copy(chunkPath, outputStream);
            }
        } catch (IOException e) {
            return Mono.error(e);
        }

        // 可选：删除临时目录
        try {
            Files.walk(tempDir)
                    .sorted(Comparator.reverseOrder())
                    .map(Path::toFile)
                    .forEach(File::delete);
        } catch (IOException ignored) {
        }

        return Mono.just("Merge complete: " + fileName);
    }


    /**
     * @param filePart
     * @description: 单文件上传
     * @return: reactor.core.publisher.Mono<org.springframework.http.ResponseEntity < java.lang.String>>
     * @author yangxiaodong
     * @date: 2025/7/2 11:05
     */
    @PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<ResponseEntity<String>> uploadFileWithProgress(@RequestPart("file") FilePart filePart) {
        String filename = filePart.filename();
        Path destPath = Paths.get(uploadDir, filename);
        // 文件大小，依赖前端传入
        long totalBytes = filePart.headers().getContentLength();
        // 上传开始时间
        Instant startTime = Instant.now();
        // 异步打开文件通道（含目录创建）
        Mono<AsynchronousFileChannel> channelMono = Mono.fromCallable(() -> {
            // 可能阻塞
            Files.createDirectories(destPath.getParent());
            return AsynchronousFileChannel.open(destPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
        }).subscribeOn(Schedulers.boundedElastic());

        return channelMono.flatMap(channel -> {
            AtomicLong written = new AtomicLong(0);
            // 每 3 秒打印一次
            UploadLogger logger = new UploadLogger(totalBytes, 3);
            return DataBufferUtils.write(filePart.content(), channel)
                    .doOnSubscribe(s -> log.info("{}:开始上传", filename))
                    .doOnNext(dataBuffer -> logger.onChunk(dataBuffer.readableByteCount()))
                    .doOnComplete(() -> {
                        log.info("{}:上传完成", filename);
                        Duration duration = Duration.between(startTime, Instant.now());
                        // 避免除以0
                        long seconds = Math.max(duration.getSeconds(), 1);
                        long total = written.get();
                        // KB/s
                        long speed = total / seconds / (1024 * 1024);
                        log.info("上传完成: {}", filename);
                        log.info("总大小: {} bytes", total);
                        log.info("耗时: {} ms", duration.toMillis());
                        log.info("平均速度: {} Mb/s", speed);
                    })
                    .publishOn(Schedulers.boundedElastic())
                    .doFinally(signal -> {
                        try {
                            channel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    })
                    .then(Mono.just(ResponseEntity.ok("上传成功: " + filename)))
                    .onErrorResume(e -> {
                        log.error("上传失败: ", e);
                        return Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                                .body("上传失败: " + e.getMessage()));
                    });
        });
    }

    /**
     * @param fileParts
     * @description: 多文件上传
     * @return: reactor.core.publisher.Mono<org.springframework.http.ResponseEntity < java.lang.String>>
     * @author yangxiaodong
     * @date: 2025/7/2 11:05
     */
    @PostMapping(value = "/upload-multiple", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<ResponseEntity<String>> uploadMultipleFiles(@RequestPart("files") Flux<FilePart> fileParts) {
        return fileParts
                .flatMap(filePart -> {
                    Path uploadPath = Paths.get(uploadDir, filePart.filename());
                    try {
                        if (!Files.exists(uploadPath.getParent())) {
                            Files.createDirectories(uploadPath.getParent());
                        }
                    } catch (IOException e) {
                        return Mono.error(new RuntimeException("创建目录失败"));
                    }
                    return filePart.transferTo(uploadPath);
                })
                .then(Mono.just(ResponseEntity.ok("所有文件上传成功")))
                .onErrorResume(e -> Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                        .body("上传失败: " + e.getMessage())));
    }

    /**
     * @param filename
     * @description: 文件下载
     * @param:
     * @return: reactor.core.publisher.Mono<org.springframework.http.ResponseEntity < org.springframework.core.io.Resource>>
     * @author yangxiaodong
     * @date: 2025/7/2 11:06
     */
    @GetMapping("/download/{filename}")
    public Mono<ResponseEntity<Flux<DataBuffer>>> download(@PathVariable String filename) {
        Path filePath = Paths.get(uploadDir).resolve(filename);
        if (!Files.exists(filePath)) {
            return Mono.just(ResponseEntity.notFound().build());
        }
        try {
            FileSystemResource resource = new FileSystemResource(filePath);
            long fileSize = resource.contentLength();

            AtomicLong startNanos = new AtomicLong();

            Flux<DataBuffer> body = DataBufferUtils.read(
                            resource,
                            new DefaultDataBufferFactory(),
                            4096
                    )
                    .doOnSubscribe(subscription -> {
                        startNanos.set(System.nanoTime());
                        log.info("开始下载:{}", filename);
                    })
                    .doOnComplete(() -> {
                        long duration = System.nanoTime() - startNanos.get();
                        double seconds = duration / 1_000_000_000.0;
                        double speed = fileSize / (1024.0 * 1024.0) / seconds;
                        log.info("下载完成:{} s，平均速度:{} Mb/s", seconds, speed);
                    });

            return Mono.just(ResponseEntity.ok()
                    .contentType(MediaType.APPLICATION_OCTET_STREAM)
                    .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + filename + "\"")
                    .body(body));

        } catch (Exception e) {
            return Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build());
        }
    }
}
